package org.hong.monkey.rdd

import org.hong.monkey.util.ClosureCleaner
import org.hong.monkey.Logging
import org.hong.monkey.annotation.Experimental
import org.hong.monkey.crawl.{CrawlResponse, CrawlRequest}

import scala.reflect.ClassTag

/**
 * Created by wangsihong on 2016/10/31.
 */

class CrawlRDDFunctions(self: RDD[CrawlRequest]) extends Serializable with Logging {

  @Experimental
  def extract[U : ClassTag](func: CrawlResponse => U): RDD[U] = self.withScope {
    ClosureCleaner.clean(func, true)
    new MapPartitionsRDD[U, CrawlRequest](self, (context, pid, iter) => iter.map( request => {
      val response = request.download()
      func(response)
    }))
  }
}

